package com.wyd.config;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.wyd.entity.Message;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.Date;

@Component
@Slf4j
public class KafkaSender {

  private final KafkaTemplate<String, String> kafkaTemplate;

  //构造器方式注入 kafkaTemplate
  public KafkaSender(KafkaTemplate<String, String> kafkaTemplate) {
    this.kafkaTemplate = kafkaTemplate;
  }

  private Gson gson = new GsonBuilder().create();

  public void send(String topicName,String msg) {
    try{
      //生产消息
      Message message = new Message();

      message.setId(System.currentTimeMillis());
      message.setMsg(msg);
      message.setSendTime(new Date());
      log.info("发送消息>>>message ：{}", gson.toJson(message));

      ListenableFuture<SendResult<String, String>> listenableFuture = kafkaTemplate.send(topicName, gson.toJson(message));
      listenableFuture.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
        @Override
        public void onSuccess(SendResult<String, String> result) {
          log.info("发送成功回调：{}",gson.toJson(result.getProducerRecord().value()));
        }
        @Override
        public void onFailure(Throwable ex) {
          log.error("发送失败，topicName：{},msg:{}",topicName,msg);
        };
      });
    }catch (Exception e){
      log.error("发送异常，topicName：{},msg:{}",topicName,msg);
      e.printStackTrace();
    }

  }

}